package com.ruoyi.ai.listener;

import com.alibaba.dashscope.aigc.generation.GenerationOutput;
import com.alibaba.dashscope.aigc.generation.GenerationResult;
import com.alibaba.dashscope.common.Message;
import com.ruoyi.ai.domain.AiQuestionLog;
import com.ruoyi.ai.service.IAiQuestionLogService;
import com.ruoyi.ai.service.MailService;
import com.ruoyi.ai.service.impl.ChatServiceImpl;
import com.ruoyi.common.rocketmqmsg.SendAiMsg;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
import lombok.SneakyThrows;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 监听ai主题并消费
 * @author zhaoYuFei
 */
@Component
@RocketMQMessageListener(topic = "ai", consumerGroup = "powernode-group",messageModel = MessageModel.CLUSTERING)
public class TYQWListener implements RocketMQListener<SendAiMsg> {
    @Autowired
    ChatServiceImpl chatService;
    @Autowired
    IAiQuestionLogService iAiQuestionLogService;
    @Autowired
    MailService mailService;

    @SneakyThrows
    @Override
    public void onMessage(SendAiMsg message) {
//        调用ai接口并返回值
        GenerationResult generationResult = chatService.chatAi(message.getMsg());
        List<GenerationOutput.Choice> choices = generationResult.getOutput().getChoices();
        List<Message> collect = choices.stream().map(GenerationOutput.Choice::getMessage).collect(Collectors.toList());
        List<AiQuestionLog> result = collect.stream().map(e -> {
            AiQuestionLog aiQuestionLog = new AiQuestionLog();
            aiQuestionLog.setContent(e.getContent());
            aiQuestionLog.setUserId(message.getUserId());
            aiQuestionLog.setQuestion(message.getMsg());
            return aiQuestionLog;
        }).collect(Collectors.toList());
//        将查询记录传入数据库
        iAiQuestionLogService.insertAiQuestionLogs(result);
//        开启线程池,将记录以mail的形式发送出去
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
        for (AiQuestionLog aiQuestionLog : result) {
            scheduledExecutorService.execute(new Runnable() {
                @Override
                public void run() {
                    mailService.sendMail(aiQuestionLog.getContent(),aiQuestionLog.getUserId().toString());
                }
            });
        }
        scheduledExecutorService.shutdown();
    }
}
